package cc.verywell.pureblock.msg.nfs;

import java.lang.ref.WeakReference;
import java.nio.channels.CompletionHandler;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import cc.verywell.pureblock.func.node.KeyShell;
import cc.verywell.pureblock.func.node.Node;
import cc.verywell.pureblock.func.node.NodeBucket;
import cc.verywell.pureblock.func.node.NodeBucket.ValueShell;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.log.ThreadListener;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadSupply;
import cc.verywell.pureblock.msg.p2p.P2pFuncProvider;
import cc.verywell.pureblock.net.AioClientImpl0;
import cc.verywell.pureblock.net.AioClientConnector;
import cc.verywell.pureblock.net.AioClientInterface;
import cc.verywell.pureblock.net.Msg;
import cc.verywell.pureblock.net.NormalSession;

/**
* 
* Copyright © 2019 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2019年4月25日 下午9:38:30
*/
public class NfsPolicyV1 implements NfsListener, NfsControler {
	static public int UPLOAD_CONNECTING_NODES = 2; //快速分发数量 
	static public int UPLOAD_TOTAL_NODES = 3; //精准分发数量
	static public int SEEK_NODES = 5; //搜寻数量
	static public long SEEK_0_TIMEOUT = 1000; //快速搜寻超时 (毫秒),已连接的目标,此时间内没回应就当作没有
	static public int TAKE_QUE = 2; //每个连接的同时下载队列;
	static public long TAKE_TIMEOUT = 3000; //下载超时 (毫秒)
	static public long TODO_TIMEOUT = 10000; //运行超时 (毫秒)
	static public long REMOTE_NEWORDERCHECK_TIMEOUT = 30000; //新到ORDER检查超时 (毫秒)

	public static boolean downloadDebuging = false;
	public static boolean newStoredDebuging = false;
	boolean testNoTakeSwitch = false;
	
	private NfsNodeBucket<PolicyNodeShell> bucket = new NfsNodeBucket<PolicyNodeShell>() {
		protected PolicyNodeShell newShell(ValueShell v) {return new PolicyNodeShell(v);};
		protected PolicyNodeShell[] newShellArray(int size) {return new PolicyNodeShell[size];};
	};
	private ConcurrentHashMap<KeyShell, PolicyGet> requesting = new ConcurrentHashMap<KeyShell, PolicyGet>();
	
	///////////这里开始是依赖的外部类
	
	private NfsFuncProvider nfs;
	private P2pFuncProvider p2p;
	public void setProvicer(NfsFuncProvider nfs,P2pFuncProvider p2p) {
		this.nfs = nfs;
		this.p2p = p2p;
	}
	
	private NodeBucket pubBucket;
	public void setBucket(NodeBucket pubBucket) {
		this.pubBucket = pubBucket;
	}
	
	private AioClientInterface<NormalSession, AioClientConnector> client;
	public void setClient(AioClientInterface<NormalSession, AioClientConnector> client) {
		this.client = client;
	}
	
	private UploadSupply nfsidxUploadIn;
	public void setUploadListenerIn(UploadSupply in) {
		this.nfsidxUploadIn = in;
	}
	
	private NfsAutoStoreRule nfsAutoStock;
	public void setNfsAutoStock(NfsAutoStoreRule auto) {
		this.nfsAutoStock = auto;
	}
	
	///////////超时管理控件，先简单搞一下，以后一定要优化的
	
	protected void seek_0_alarm(final long timeAdd,PolicyGet apn) {
		if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(apn.hashCode())," seted first seek alarm:",timeAdd);
		alarm(timeAdd,pn->{
			if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," first seek alarm wake, remoteList:",null==pn.takingList?"NO":pn.takingList.size());
			seek_0_timeout(pn);
		},apn);
	}
	protected void taking_alarm(final long timeAdd,PolicyGet apn,final PolicyNodeShell ps) {
		if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(apn.hashCode())," seted download alarm:",timeAdd);
		alarm(timeAdd,pn->{
			if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," download alarm wake , remoteList:",null==pn.takingList?"NO":pn.takingList.size(),", binding remote ",ps.v," , and it's takingList:",null==ps.takingList?"NO":ps.takingList.size());
			fireTaked(ps.v, pn.key, false, true); //TODO 还要继续给node分配任务吗?
			//if(debuging&&Pblog.DEBUG)Pblog.DEBUG(Integer.toHexString(pn.hashCode())," take_alarm_end pg:",null==pn.takingList?"NO":pn.takingList.size()," ps:",null==ps.takingList?"NO":ps.takingList.size());
		},apn);
	}
	protected void expire_alarm(final long timeAdd,PolicyGet apn) {
		if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(apn.hashCode())," seted expire alarm:",timeAdd);
		alarm(timeAdd,pn->{
			if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," expire alarm wake, remoteList:",null==pn.takingList?"NO":pn.takingList.size());
			long extendTimeAdd = 0;
			synchronized (pn) {
				long time = System.currentTimeMillis();
				if(time>=pn.startTime && time<pn.todoExpire) {//没有发生调整，而且时间还没到;
					extendTimeAdd = pn.todoExpire-time;
				}
			}
			if(extendTimeAdd>0) {
				expire_alarm(extendTimeAdd,pn); //延期;
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," expire alarm extended: ",extendTimeAdd);
				return;
			}
			if(requesting.remove(pn.key,pn)) {
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," removed cause timeout",pn.isdone?" and done.":"");
				KeyShell key = pn.key;
				byte[] payload = null;
				{
					NfsFuncProvider nfs = this.nfs;
					if(nfs!=null) {
						NfsLocalStoreProvider fsmap = nfs.fsmap;
						if(fsmap!=null)
							payload = fsmap.get(key);
					}
				}
				PolicyNodeShell ps = null;
				synchronized (pn) {
					ps = pn.takingRemote;
					if(ps!=null) {
						ps.dowing.decrementAndGet();
						pn.takingRemote = null;
						pn.takingList.remove(ps.v.key);
					}
					pn.isdone = true;
					if(pn.todolist!=null) {
						try {
							for (BiConsumer<KeyShell, byte[]> bi : pn.todolist) {
								bi.accept(key, payload);
							}
						}finally {
							pn.todolist.clear();
						}
					}
				}
				if(ps!=null) {
					toTake(ps);
				}
//				if(debuging&&Pblog.DEBUG)
//					Pblog.DEBUG(Integer.toHexString(pn.hashCode())," expire_alarm_finished done:",pn.isdone," takingList:",null==pn.takingList?"NO":pn.takingList.size());
			}else
				if(downloadDebuging&&Pblog.DEBUG)
					Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," already removed",pn.isdone?" and done.":", but not done.");
		},apn);
	}
	protected void alarm(long timeAdd,Consumer<PolicyGet> run,PolicyGet pn) {
		synchronized(alarmList) {
			if(alarmClosed) return;
			long time = System.currentTimeMillis();
			alarmFixTime(time);
			
			time += timeAdd;
			alarmList.add(new AlarmRun(time,run,pn));
			if(time<alarmMinTime) {
				alarmList.notify();
			}
		}
	}
	@Override
	public void close() {
		synchronized(alarmList) {
			if(alarmClosed) return;
			alarmList.clear();
			alarmClosed = true;
			alarmList.notify();
		}
	}
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	private ThreadListener tl;
	public NfsPolicyV1(ThreadGroup tg,ThreadListener threadListener) {
		this.tl = threadListener!=null?threadListener:ThreadListener.NONE;
		Thread t = new Thread(tg, ()->{
			tl.threadIn();
			cycle:for(;;) {
				LinkedList<AlarmRun> doList = null;
				synchronized(alarmList) {
					if(alarmClosed) break cycle;
					long time = System.currentTimeMillis();
					alarmFixTime(time);
					
					alarmMinTime = Long.MAX_VALUE;
					Iterator<AlarmRun> it = alarmList.iterator();
					while(it.hasNext()) {
						AlarmRun a = it.next();
						if(a.time()<=time) {
							if(doList==null)doList = new LinkedList<AlarmRun>();
							doList.add(a);
							it.remove();
						}else {
							alarmMinTime = Math.min(alarmMinTime, a.time());
						}
					}
					if(doList==null) {
						tl.threadWait();
						try {
							alarmList.wait(alarmMinTime-time);
						} catch (InterruptedException e) {
							if(Pblog.FS)Pblog.FS(e);
						}
						tl.threadWake();
						continue;
					}
				}
				for (AlarmRun alarmRun : doList) {
					alarmRun.run();
				}
			}
			tl.threadOut();
		},"NfsPolicyV1_alarm");
		t.setPriority(t.getPriority()-1);
		t.start();
	}
	final private LinkedList<AlarmRun> alarmList = new LinkedList<AlarmRun>();
	private long alarmMinTime = 0;
	private boolean alarmClosed;
	private long alarmLastSync;
	private void alarmFixTime(long time) { //必须在 synchronized(alarmList) 块内使用;
		if(time<alarmLastSync) { //时间被调整了；
			long offset = time-alarmLastSync;
			for (AlarmRun alarmRun : alarmList) {
				alarmRun.timeAdd(offset);
			}
			if(Pblog.FS)Pblog.FS("nfs alarm time changed offset: ",offset);
		}
		alarmLastSync = time;
	}
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	static class AlarmRun extends WeakReference<NfsPolicyV1.PolicyGet>{
		private long time = 0;
		private Consumer<PolicyGet> c;
		public AlarmRun(long time, Consumer<PolicyGet> c ,PolicyGet g) {
			super(g);
			this.time = time;
			this.c = c;
		}
		public long time() {
			return null==super.get()?0:this.time;
		}
		public void timeAdd(long t) {
			this.time += t;
		}
		public void run() {
			PolicyGet pg = super.get();
			if(pg!=null)
				this.c.accept(pg);
		}
	}
	
	//////////检索核心
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	@Override
	public void newSeeking(BiConsumer<KeyShell, byte[]> func, long timeoutAdd, KeyShell... filehash) {//搜索文件 - 步骤0;
		if(timeoutAdd==0)timeoutAdd = TODO_TIMEOUT;
		long timeStart = System.currentTimeMillis();
		NfsFuncProvider provider = this.nfs;
		if(provider==null)return;
		for (KeyShell f : filehash) {
			PolicyGet pn = new PolicyGet(f);
			pn.startTime = timeStart;
			pn.todoExpire = timeStart+timeoutAdd;
			pn.seekRound = 0; //正在计算K值
			if(func!=null){
				pn.todolist = new LinkedList<>();
				pn.todolist.add(func);
			}
			{
				PolicyGet old = requesting.putIfAbsent(pn.key, pn);
				if(old!=null) {
					pn = old;
					synchronized (pn) {
						if(func!=null){
							if(pn.todolist==null) pn.todolist = new LinkedList<>();
							pn.todolist.add(func);
						}
						if(pn.startTime>timeStart) { //时间发生了调整;
							pn.startTime = timeStart;
							pn.todoExpire = timeStart+timeoutAdd;
						}else if(pn.todoExpire<timeStart+timeoutAdd) {//时间延期;
							pn.startTime = timeStart; //不断延期后又发生时间往前调整，就需要用startTime到todoExpire之间锁定运行区间;
							pn.todoExpire = timeStart+timeoutAdd;
						}
					}
					if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," init join todo , timeout: +",timeoutAdd);
					continue;
				}
			}
			///以上是旧有添加，接下来真正开始新建
			expire_alarm(timeoutAdd,pn);
			NfsFuncProvider.SeekFileMsg m = provider.new SeekFileMsg(Collections.singletonList(f.key));
			{//已连接搜索
				List<PolicyNodeShell> nodes = bucket.getTargetNearby(f.key, SEEK_NODES);
				//HashSet<KeyShell> set = new HashSet<KeyShell>(SEEK_NODES);
				int distance = -1;
				boolean success=false;
				for (PolicyNodeShell s : nodes) {
					NormalSession ns = s.v.getSession();
					if(null!=ns && ns.writeMsg(m)) { //其实这时write并不能保证一定发出去了;
						success=true;//set.add(s.v.key);
						int d = s.v.getDistance(f.key);
						if(d>distance)distance=d;
						if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," seeking file from session ",ns," distance: ",d);
					}
				}
				if(distance<=0)distance = pubBucket.getMyNode().getDistance(f.key);
				synchronized (pn) {//策略V1
//					if(pn.seekedList==null)
//						pn.seekedList = set;
//					else
//						pn.seekedList.addAll(set);
					pn.seekDistance = distance;
				}
//				Pblog.print("//////////////////////////",Integer.toHexString(pn.hashCode()), " success:",success);
				if(success/*set.size()>0*/) {
					seek_0_alarm(SEEK_0_TIMEOUT, pn);
				}else {
					seek_0_timeout(pn);
				}
			}
		}
	}
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	protected void seek_0_timeout(PolicyGet pn) { //搜索文件 - 步骤1;
		//if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG(Integer.toHexString(pn.hashCode())," seek_alarm_timeout seek0:",pn.isSeek0()," seeking:",pn.isSeeking()," seekAlarmed:",pn.seekAlarmed);
		Set<KeyShell> waiting; 
		synchronized (pn) {
			pn.seekAlarmed = true;
			if(!pn.isSeek0())return;
			pn.seekRound=1;
			pn.seekDistanceWaiting = waiting = Collections.synchronizedSet(new HashSet<>());
		}
		//if(debuging&&Pblog.DEBUG)Pblog.DEBUG(Integer.toHexString(pn.hashCode())," seek_alarm_init");
		//在getTargetNearby之前新加入的node会在getTargetNearby处理;
		byte[] f = pn.key.key;
		P2pFuncProvider p2p = this.p2p;
		if(p2p==null)return;
		P2pFuncProvider.PeersWantMsg m = p2p.new PeersWantMsg(f);
		List<PolicyNodeShell> nodes = bucket.getTargetNearby(f, SEEK_NODES);//新加进来的都已经request过了;
		int distance = -1;
		int successCount = 0;
		for (PolicyNodeShell s : nodes) {
			NormalSession ns = s.v.getSession();
			if(waiting.remove(s.v.key) || (null!=ns && ns.getFunc().hasFunc(p2p) && ns.writeMsg(m)) ) {//其实这时write并不能保证一定发出去了;
				int d = s.v.getDistance(f);
				if(d>distance)distance = d;
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," seeking peer from session ",ns," distance: ",d);
				successCount++;
			}
		}
		//在getTargetNearby之后新加入的node在seekDistanceWaiting;
		synchronized (pn) {
			if(distance>=0)
				pn.seekDistance = distance;
			pn.seekDistanceWaiting = null;
		}
		if(0==successCount)if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," cannot seeking file from any session ");
		NodeBucket pubBucket = this.pubBucket;
		if(pubBucket!=null) { // 远程
			List<ValueShell> publs = pubBucket.getTargetNearby(f, SEEK_NODES);
			for (ValueShell v : publs) {
				String[] remote = v.getRemote();
				{//评估是否连接
					if(!ValueShell.isConnectable(remote))continue;
					NormalSession session = v.getSession();
					if(session!=null && !session.isClosed())continue;
				}
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," connecting new node ",v,", address: ",v.getRemote());
				client.connect(v, null, CH_NULL);
			}
		}
	}
	
	/////检索核心 之 新到node策略
	static private CompletionHandler<NormalSession, Void> CH_NULL = new CompletionHandler<NormalSession, Void>() {
		@Override
		public void completed(NormalSession result, Void attachment) {}
		@Override
		public void failed(Throwable exc, Void attachment) {}
	};
	
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	class NewNodePredicate implements Predicate<ValueShell>{
		String nfs;
		NewNodePredicate(){
			NfsFuncProvider nfs = NfsPolicyV1.this.nfs;
			if(nfs!=null)this.nfs = nfs.funcName();
//			P2pFuncProvider p2p = NfsPolicyV1.this.p2p;
//			if(p2p!=null)this.p2p = p2p.funcName();
		}
		@Override
		public boolean test(ValueShell v) {
			NormalSession ns = v.getSession();
			if(ns!=null && !ns.isClosed())return false;
			Node n = v.getNode();
			if(n==null)return false;
			String[] t = n.getFuncs();
			if(t==null)return false;
			if(nfs!=null) {
				for (String s : t) {
					if(nfs.equals(s))return true;
				}
			}
//			if(p2p!=null) {
//				for (String s : t) {
//					if(p2p.equals(s))return true;
//				}
//			}
			return false;
		}
	}
	@Override
	public void fireNewNodeKey(Collection<ValueShell> values) {
		AioClientInterface<NormalSession, AioClientConnector> client = this.client;
		if(client==null)return;
		//Map<KeyShell,ValueShell> keys = values.stream().filter(new NewNodePredicate()).map(a->a.key).collect(Collectors.toMap());
		HashMap<KeyShell,ValueShell> keys = new HashMap<KeyShell,ValueShell>();
		{
			NewNodePredicate test = new NewNodePredicate();
			for (ValueShell v : values) {
				if(test.test(v))keys.put(v.key, v);
			}
		}
		for (PolicyGet pn : requesting.values()) {
			synchronized (pn) {
				if(pn.seekRound!=0 && pn.isSeeking()) {
					Iterator<Entry<KeyShell,ValueShell>>it = keys.entrySet().iterator();
					while(it.hasNext()) {
						Entry<KeyShell,ValueShell> newentry = it.next();
						KeyShell newk = newentry.getKey();
						int d;
						if((d=NodeBucket.distance(pn.key.key, newk.key))<=pn.seekDistance) {
							it.remove();
							if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," will connect new node ",newentry.getValue(),", distance: ",d,"<=",pn.seekDistance);
						}
					}
				}
			}
		}
		
		for (ValueShell v : values) {
			String[] remote = v.getRemote();
			{//评估是否连接
				if(!ValueShell.isConnectable(remote))continue;
				if(keys.containsKey(v.key))continue;
				NormalSession session = v.getSession();
				if(session!=null && !session.isClosed())continue;
			}
			if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: [] connecting new node ",v,", address: ",v.getRemote());
			client.connect(v, null, CH_NULL);
//			for (String remotestr : remote) {
//				InetSocketAddress address = StringUtil.stringToPort(remotestr);
//				try {
//					Future<NormalSession> to = client.connect(address);
//					break;
//				} catch (IOException e) {
//					if(Pblog.DEBUG)Pblog.DEBUG(e);
//				}
//			}
		}
	}
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	@Override
	public void fireNewSession(NormalSession nc) {
		//找文件
		Set<KeyShell> set = requesting.keySet();
		{
			ArrayList<byte[]> al = new ArrayList<>(set.size());
			for (KeyShell k : set) {
				al.add(k.key);
			}
			List<Msg> ml = nfs.getSeekFileMsg(al);
			if(ml!=null) {
				for (Msg msg : ml) {
					nc.writeMsg(msg);
				}
			}
		}
		//找文件相关的节点;
		P2pFuncProvider p2p = this.p2p;
		if(p2p!=null && nc.getFunc().hasFunc(p2p)) {
			ValueShell node = nc.getFunc().getNode();
			for (PolicyGet pn : requesting.values()) {
				synchronized (pn) {
					int d;
					if(pn.seekRound!=0 && pn.isSeeking() && (d=node.getDistance(pn.key.key))<=pn.seekDistance) {
						if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," seek peer from new session ",nc," distance: ",d,"<=",pn.seekDistance);
						P2pFuncProvider.PeersWantMsg m = p2p.new PeersWantMsg(pn.key.key);
						if(nc.writeMsg(m) && pn.seekDistanceWaiting!=null) {
							pn.seekDistanceWaiting.add(node.key);
						}
					}
				}
			}
		}
		//最后添加
		bucket.addValue(nc.getFunc().getNode());
		//发送seek列表
	}

	@Override
	public void fireCloseSession(NormalSession nc) {
		bucket.removeValue(nc.getFunc().getNode());
	}

	
	////////////////下载核心
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	@Override
	public void fireStoreFile(NormalSession nc, List<byte[]> fileHash) { //同一node同一文件可能会发送多次;
		NfsFuncProvider nfs = this.nfs;
		if(nfs==null || nfs.fsmap==null)return ;
		PolicyNodeShell ps = bucket.getValue(nc.getFunc().getNode().key);
		if(ps==null) {
			if(Pblog.FS)Pblog.FS("policyV1 fire stored at unknow key ",nc.getFunc().getNode().key);
		}else {
			long startTime = System.currentTimeMillis();
			for (byte[] f : fileHash) {
				KeyShell key = new KeyShell(f);
				if(nfs.fsmap.contain(key))continue;
				PolicyGet pn = new PolicyGet(key);
				pn.startTime = startTime;
				pn.takingList = new LinkedHashMap<>();
				pn.takingList.put(ps.v.key,ps);
				boolean psaddTaking = true;
				PolicyGet old = requesting.putIfAbsent(pn.key, pn);
				if(old!=null) {
					synchronized (old) {
						if(old.isdone) {
							continue;
						}else {
							if(old.takingList==null)
								old.takingList=pn.takingList;
							else
								psaddTaking = old.takingList.put(ps.v.key,ps)!=ps;
						}
					}
					pn = old;
					if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," will take from remote ",nc);
				}else {
					if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," will take from remote ",nc,", though we don't need it." );
				}
				if(psaddTaking)
					ps.addTaking(pn);
			}
			toTake(ps);
		}
		if(nfsidxUploadIn!=null)nfsidxUploadIn.fireStoreFile(nc, fileHash);
	}
	protected void toTake(PolicyNodeShell ps) {
		toTake(ps,ps.dowing.get());
	}
	protected void toTake(PolicyNodeShell ps,int downing) {
		if(downing<0) {
			if(Pblog.FS)Pblog.FS("downing==",downing," at ",ps.v.key.key);
			downing=1;
		}
		NfsFuncProvider nfs = this.nfs;
		if(nfs==null || nfs.fsmap==null)return ;
		while(downing<TAKE_QUE) {
			PolicyGet pn = ps.getTaking();
			if(pn==null)return;
			if(nfs.fsmap.contain(pn.key)) {
				fireTaked(ps.v, pn.key, true, false);
			}
			NormalSession ns = ps.v.getSession();
			if(ns==null || !ns.writeMsg(nfs.new WantFileMsg(pn.key.key))) {
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," cannot take from remote ",ps.v );
				fireTaked(ps.v, pn.key, false, false);
			}else {
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," taking from remote ",ps.v );
				downing++;
				taking_alarm(TAKE_TIMEOUT, pn, ps);
			}
		}
	}
	
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	protected void fireTaked(ValueShell knode, KeyShell fileHash, boolean isMarkDone, boolean toTake) {
		PolicyGet pn = requesting.get(fileHash);
		if(pn!=null) {
			PolicyNodeShell ps = null;
			int dowing = 0;
			boolean toSeek0 = false;
			List<PolicyNodeShell> freePNS = null;
			synchronized (pn) {
				ps = pn.takingRemote;
				if(ps!=null) {
					if(knode.key.equals(ps.v.key)) {
						if(isMarkDone) {
							pn.isdone = true; // 在这里标记完结，以防继续下载;
							if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," done from ",knode);
						}else {
							pn.takingList.remove(ps.v.key); //否则就是当前连接有问题，需要移除;
							if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," remote kick out ",knode," , it may has some wrong or delay.");
						}
						dowing = ps.dowing.decrementAndGet();
						pn.takingRemote = null;
						if(pn.isSeek0() && pn.seekAlarmed) {// 继续seek;
							toSeek0 = true;
						}else if(!pn.isdone){//查找pn的空闲node，让他们跑起来;
							freePNS = pn.takingList.values().stream().filter(a->a.dowing.get()==0).collect(Collectors.toList());
						}
					}else {
						ps=null;
					}
				}
			}
			if(toSeek0)seek_0_timeout(pn);
			if(ps!=null && toTake) {
				toTake(ps, dowing);
			}
			if(freePNS!=null) {
				freePNS.forEach(a->toTake(a));
			}
		}
	}
	
	
	//////////上传核心
	
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	
	@Override
	public void newStored(Map<KeyShell, Set<KeyShell>> editAbleFromMap) {//分发文件,要去掉已在列表内的
		NfsFuncProvider provider = this.nfs;
		if(provider==null)return;
		//for (KeyShell f : knownFromMap.keySet()) {
		//	//以后要改成汇总再分发;
		//	newStoredAfterTake(f,null);//现在newStore只对外 newStoredAfterTake(f,requesting.get(new KeyShell(f)));
		//}
		
		//20200206已重新编译
		if(editAbleFromMap==null || editAbleFromMap.isEmpty())return;
		if(editAbleFromMap.size()==1) {
			//最简单的处理办法
			editAbleFromMap.forEach((k,v)->newStoredAfterTake(k,v));
			return;
		}
		
		{//已连接写入
			final Map<KeyShell, StoredPrepareNode> remote = new ConcurrentHashMap<>();//remote为多线程操作
			////数据汇总
			try(
					Stream<Entry<KeyShell, Set<KeyShell>>> stream = editAbleFromMap.entrySet().parallelStream();
				) {
				stream.forEach(entry->{ //并行处理
					Set<KeyShell> writed = entry.getValue();
					List<PolicyNodeShell> nodes = bucket.getTargetNearby(entry.getKey().key, UPLOAD_CONNECTING_NODES);
					for (PolicyNodeShell s : nodes) {
						KeyShell remoteNode = s.v.key;
						if(writed!=null && writed.contains(remoteNode)) continue;
						
						StoredPrepareNode newsp = new StoredPrepareNode();
						StoredPrepareNode sp = remote.putIfAbsent(remoteNode, newsp);
						if(sp==null)sp = newsp;

						sp.add(s, entry); //remote为多线程操作，需要sync;
					}
				});
			}
			////数据分发
			try(
					Stream<StoredPrepareNode> stream = remote.values().parallelStream();
				) {
				stream.forEach(sp->{ //并行处理
					PolicyNodeShell pn = sp.policyNode();
					NormalSession ns = pn.v.getSession();
					if(null!=ns) { //不管是否已关闭;
						KeyShell remoteNode = pn.v.key;
						if(newStoredDebuging&&Pblog.DEBUG)Pblog.DEBUG("send newStoreConnect:",sp.keySize()," at ",ns," detail: ",sp.keyShow());
						NfsFuncProvider.StoreFileMsg m = provider.new StoreFileMsg(sp.keyList());
						ns.writeMsg(m);
						sp.joinRemoteNode(remoteNode);//登记已发送node
					}
				});
			}
		}
		//未连接写入
		if(UPLOAD_TOTAL_NODES>0) {
			NodeBucket pubBucket = this.pubBucket;
			if(pubBucket!=null) { //  异步+汇总
				//与 NfsAutoStoreRemoteHandler 配合使用
				final NfsPolicyV1newStoreHandler asrh = new NfsPolicyV1newStoreHandler();
				try(
						Stream<Entry<KeyShell, Set<KeyShell>>> stream = editAbleFromMap.entrySet().parallelStream();
					) {
					stream.forEach(entry->{ //并行处理
						Set<KeyShell> writed = entry.getValue();
						
						LinkedList<NodeBucket.ValueShell> list = new LinkedList<>();
						int okrest = UPLOAD_TOTAL_NODES;
						{
							int i = 0;
							List<NodeBucket.ValueShell> nodes =  pubBucket.getTargetNearby(entry.getKey().key, UPLOAD_TOTAL_NODES*2);
							for (ValueShell v : nodes) {
								if(!ValueShell.isConnectable(v.getRemote()))continue;
								i++;
								if(writed!=null && writed.contains(v.key)) {
									if(i<=UPLOAD_CONNECTING_NODES) {
										okrest--;
										if(okrest==0 && list.isEmpty())
											break;
									}
									continue;
								}
								list.add(v);
							}
						}
						if(!list.isEmpty()) {
							asrh.add(entry.getKey(), list, okrest); //remote为多线程操作，需要sync;
						}
					});
				}
				asrh.startConnect(provider, NfsPolicyV1.this.client); //开始执行;
				
			}
		}
	}
	static private final class StoredPrepareNode{//已连接写入专属类
		private PolicyNodeShell policyNode;
		private HashMap<KeyShell, Entry<KeyShell, Set<KeyShell>>> keyMap;
		///在多线程中调用(汇总线程);
		synchronized public void add(PolicyNodeShell node,Entry<KeyShell, Set<KeyShell>> entry) { 
			this.policyNode = node;
			if(keyMap==null) {
				keyMap = new HashMap<>();
			}
			keyMap.put(entry.getKey(), entry);
		}
		///在独立线程中调用(分发线程);
		public PolicyNodeShell policyNode() {return policyNode;}
		public List<byte[]> keyList() {
			try(
					Stream<KeyShell> steam = keyMap.keySet().stream();
				) {
				return steam.map(k->k.key).collect(Collectors.toList());
			}
		}
		public void joinRemoteNode(final KeyShell remoteNode) {
			try(
					Stream<Entry<KeyShell, Set<KeyShell>>> steam = keyMap.values().parallelStream();
				) {
				steam.forEach(entry->{ //并行
					synchronized (entry) {
						Set<KeyShell> writed = entry.getValue();
						if(writed==null) {
							writed = new HashSet<KeyShell>();
							entry.setValue(writed);
						}
						writed.add(remoteNode);
					}
				});
			}
		}
		int keySize() {return keyMap.size();} 
		String keyShow() {
			StringBuffer sb = new StringBuffer();
			for (KeyShell k : keyMap.keySet()) {
				if(sb.length()==0)sb.append("[");
				else sb.append(", ");
				sb.append(Integer.toHexString(k.hashCode()));
			}
			if(sb.length()==0)sb.append("[]");
			else sb.append("]");
			return sb.toString();
		}
	}

	/**
	 * newStored 的单文件例程
	 * @param 文件key
	 * @param 一个可编辑的来源列表writed ， 可以为空; 
	 */
	private void newStoredAfterTake(KeyShell f,Set<KeyShell> writed) {
		NfsFuncProvider provider = this.nfs;
		if(provider==null)return;

		NfsFuncProvider.StoreFileMsg m = provider.new StoreFileMsg(Collections.singletonList(f.key));
		{//已连接写入
			List<PolicyNodeShell> nodes = bucket.getTargetNearby(f.key, UPLOAD_CONNECTING_NODES);
			if(writed==null)writed = new HashSet<>(nodes.size()+UPLOAD_TOTAL_NODES);
			for (PolicyNodeShell s : nodes) {
				if(writed.contains(s.v.key)) continue;
				NormalSession ns = s.v.getSession();
				if(null!=ns) { //不管是否已关闭;
					if(newStoredDebuging&&Pblog.DEBUG)Pblog.DEBUG("send newStoreConnect:.",1,". at ",ns," detail: ",Integer.toHexString(f.hashCode()));
					ns.writeMsg(m);
					writed.add(s.v.key);
				}
			}
		}
		//未连接写入
		if(UPLOAD_TOTAL_NODES>0) {
			NodeBucket pubBucket = this.pubBucket;
			if(pubBucket!=null) { //  异步+汇总，现在这样不行(20200206已在newStored实现汇总后的功能)
				LinkedList<NodeBucket.ValueShell> list = new LinkedList<>();
				int okrest = UPLOAD_TOTAL_NODES;
				{
					int i = 0;
					List<NodeBucket.ValueShell> nodes =  pubBucket.getTargetNearby(f.key, UPLOAD_TOTAL_NODES*2);
					for (ValueShell v : nodes) {
						if(!ValueShell.isConnectable(v.getRemote()))continue;
						i++;
						if(writed.contains(v.key)) {
							if(i<=UPLOAD_CONNECTING_NODES) {
								okrest--;
								if(okrest==0 && list.isEmpty())
									break;
							}
							continue;
						}
						list.add(v);
					}
				}
				if(!list.isEmpty()) {
					NewStoreRemotePack p = new NewStoreRemotePack(list, okrest, f);
					newStoreRemoteHandler.failed(null, p);
				}
			}
		}
	}
	static private class NewStoreRemotePack{
		public LinkedList<ValueShell> list;
		public int okrest;
		public KeyShell filehash;
		public NewStoreRemotePack(LinkedList<ValueShell> list, int okrest, KeyShell filehash) {
			this.list = list;
			this.okrest = okrest;
			this.filehash = filehash;
		}
	}
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	private CompletionHandler<NormalSession, NewStoreRemotePack> newStoreRemoteHandler = new CompletionHandler<NormalSession, NewStoreRemotePack>(){
		@Override
		public void completed(NormalSession n, NewStoreRemotePack p) {
			writeMsg(n, p);
			failed(null,p);
		}
		private void writeMsg(NormalSession n, NewStoreRemotePack p) {
			NfsFuncProvider nfs = NfsPolicyV1.this.nfs;
			if(nfs==null || nfs.fsmap==null)return ;
			if(!nfs.fsmap.contain(p.filehash))return;
			
			if(n!=null && n.getFunc().hasFunc(nfs)) {
				if(newStoredDebuging&&Pblog.DEBUG)Pblog.DEBUG("send newStore:",1," at ",n," detail: ",Integer.toHexString(p.filehash.hashCode()));
				NfsFuncProvider.StoreFileMsg m = nfs.new StoreFileMsg(Collections.singletonList(p.filehash.key));
				n.writeMsg(m);
				p.okrest--;
			}
		}
		@Override
		public void failed(Throwable exc, NewStoreRemotePack p) {
			while(p.okrest>0 && !p.list.isEmpty()) {
				ValueShell v = p.list.removeFirst();
				NormalSession ns = v.getSession();
				if(null!=ns && !ns.isClosed()) {
					writeMsg(ns, p);
					continue;
				}
				String[] remote = v.getRemote();
				if(!ValueShell.isConnectable(remote))continue;
				AioClientInterface<NormalSession, AioClientConnector> client = NfsPolicyV1.this.client;
				if(client==null)return;
				client.connect(v, p, this);
				return;
			}
		}
	};
	
	//远程文件添加
	@Override
	public void fireNewFileStored(NormalSession nc, KeyShell fileHash, byte[] payload) {
		boolean isHandled = false;
		PolicyGet pn = requesting.remove(fileHash);
		HashSet<KeyShell> writed = null ;
		if(pn!=null) {
			if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," got data from ",nc);
			synchronized (pn) {
				pn.isdone = true;
				if(pn.todolist!=null) {
					try {
						for (BiConsumer<KeyShell, byte[]> bi : pn.todolist) {
							bi.accept(fileHash, payload);
							isHandled = true;
						}
					}finally {
						pn.todolist.clear();
					}
				}
				//已知来源;
				if(pn.takingList!=null && !pn.takingList.isEmpty()) {
					writed = new HashSet<>(pn.takingList.keySet());
				}
			}
		}
		if(nfsAutoStock!=null)nfsAutoStock.fireNewFileStored(fileHash, payload, writed, isHandled);
		//未保存如何分发文件?else newStoredAfterTake(fileHash,writed);//分发文件;
	}
	
	//本地文件添加
	@Override
	public BiConsumer<KeyShell, byte[]> fireLocalFileAdded(KeyShell fileHash) {
		if(requesting.containsKey(fileHash)) {
			return fireLocalFileAddedCallback;
		}
		return null;
	}
	private BiConsumer<KeyShell, byte[]> fireLocalFileAddedCallback = new BiConsumer<KeyShell, byte[]>(){
		public void accept(KeyShell fileHash, byte[] payload) {
			PolicyGet pn = requesting.remove(fileHash);
			if(pn!=null) {
				if(downloadDebuging&&Pblog.DEBUG)Pblog.DEBUG("pn: ",Integer.toHexString(pn.hashCode())," got data from local");
				synchronized (pn) {
					pn.isdone = true;
					if(pn.todolist!=null) {
						try {
							for (BiConsumer<KeyShell, byte[]> bi : pn.todolist) {
								bi.accept(fileHash, payload);
							}
						}finally {
							pn.todolist.clear();
						}
					}
				}
			}
		}
	};
	
	////事件连接
	
	@Override
	public void fireMissFile(NormalSession nc, KeyShell fileHash) {
		fireTaked(nc.getFunc().getNode(), fileHash, false, true);
		// 未加惩罚措施
	}

	@Override
	public boolean fireTakeFileSuccess(NormalSession nc, KeyShell fileHash) { //如需不接受此文件请返回false, 比如是对方硬塞过来的;
		fireTaked(nc.getFunc().getNode(), fileHash, true, true);
		//未加奖励措施
		return true;
	}
	@Override
	public void fireTakeFileFaild(NormalSession nc, KeyShell fileHash) {
		fireTaked(nc.getFunc().getNode(), fileHash, false, true);
		// 未加惩罚措施
	}
	
	////对方请求文件，如需block请返回false;
	@Override
	public boolean fireWantFile(NormalSession nc, byte[] fileHash) {
		if(nfsidxUploadIn!=null)nfsidxUploadIn.fireWantFile(nc, fileHash);
		if(testNoTakeSwitch)return false;
		return true;
	}
	////对方搜索文件，如需block请返回false;
	@Override
	public boolean fireSeekFile(NormalSession nc, List<byte[]> fileHash) {
		return true;
	}

	
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	static private class PolicyNodeShell extends NfsNodeShell{
		public PolicyNodeShell(ValueShell v) {
			super(v);
		}
		private List<WeakReference<PolicyGet>> takingList;
		public synchronized void addTaking(PolicyGet... node) {
			if(takingList==null)takingList = new LinkedList<>();
			if(node!=null) {
				for (PolicyGet pg : node) {
					takingList.add(new WeakReference<NfsPolicyV1.PolicyGet>(pg));
				}
			}
		}
		public synchronized PolicyGet getTaking() { //下载顺序策略;
			if(takingList==null)return null;
			Collections.sort(takingList,(a,b)->{PolicyGet aa=a.get(),bb=b.get();
				if(aa==null)return bb==null?0:-1;
				if(bb==null)return 1;
				return aa.compareTo(bb);
			});
			Iterator<WeakReference<PolicyGet>> it = takingList.iterator();
			while(it.hasNext()) {
				PolicyGet pn = it.next().get();
				if(pn==null) {
					it.remove();
					continue;
				}
				synchronized (pn) {
					if(pn.isTaking()) {
						if(pn.takingRemote==null) {
							it.remove();
							pn.takingRemote = this;
							dowing.getAndIncrement();
							return pn;
						}
					}else {
						it.remove();
					}
				}
			}
			return null;
		}
		public AtomicInteger dowing = new AtomicInteger();
	}
	
	/**
	 * @author TanYaqiu 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	static private class PolicyGet implements Comparable<PolicyGet>{
		final KeyShell key;
		public PolicyGet(KeyShell key) {
			this.key = key;
		}
		//long seekingExpire; //stat状态  -2未开始 -1正在计算K桶 0没找到 n超时时间
		boolean seekAlarmed = false;
		int seekRound; //轮次，第0轮是找已连接的，第1-6轮开始连接新NODE;
		Set<KeyShell> seekDistanceWaiting = null;//仅当进行2-1运算时使用;
		int seekDistance;
		boolean isSeek0() {
			return this.seekRound==0 && isSeeking();
			//其它原因已经进入下一轮
			//已进入take,放弃seek
		}
		boolean isSeeking() {
			return !isTaking() && !isdone;
		}
		boolean isTaking() {
			return this.takingList!=null && !this.takingList.isEmpty() && !isdone;
		}
		Map<KeyShell,PolicyNodeShell> takingList; //可take的列表;
		
		PolicyNodeShell takingRemote; //请求远端;
		List<BiConsumer<KeyShell, byte[]>> todolist; //完成该做的事;
		long startTime; //开始时间
		long todoExpire;
		boolean isdone = false;
		
		
		/////////////
		@Override
		public int compareTo(PolicyGet o) { ////下载优先顺序;
			int mytc,mytodo;
			synchronized (this) {
				mytc = takingList==null?0:takingList.size();
				mytodo = todolist==null?0:todolist.size();
			}
			int othertc,othertodo;
			synchronized (o) {
				othertc = o.takingList==null?0:o.takingList.size();
				othertodo = o.todolist==null?0:o.todolist.size();
			}
			if(mytodo!=othertodo)return othertodo-mytodo; //需求大的优先
			if(mytc!=othertc)return mytc-othertc; //种小的优先
			return Long.compare(todoExpire, o.todoExpire); //快到期的优先
		}
		
		@Override
		public int hashCode() {
			return key.hashCode();
		}
	}
	
	@Override
	public String toString() {
		StringBuffer sb = new StringBuffer();
		sb.append(getClass().getSimpleName()).append(' ');
		sb.append("requesting:").append(requesting.size()).append(" ");
		sb.append("nfsBucket:").append(bucket.map.size());
		
		return sb.toString();
	}
	public int getRequestingSize() {
		return requesting.size();
	}
	public int getBucketSize() {
		return bucket.map.size();
	}
	
}
